package com.stqf.scala.spark;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.nio.file.Paths;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * Created By admin
 * Date: 2020/5/8
 * Description:
 */
public class WordCount {

    private static final Logger log = LoggerFactory.getLogger(WordCount.class);

    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("WordCountApp").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);

        String filename = Paths.get(System.getProperty("user.dir"), "refer", "dbscript", "softly-1.0-cham.sql").toString();

        JavaRDD<String> lines = sc.textFile(filename);

        JavaRDD<String> words = lines.filter((Function<String, Boolean>) s -> Objects.nonNull(s) && s.trim().length() > 0)
                .flatMap((FlatMapFunction<String, String>) s -> {
                    String[] wordArr = s.split("[,\"<>(){}.\\s\\[\\]\\\\:;@]+");
                    List<String> words1 = Stream.of(wordArr)
                            .filter(item -> item.matches("[a-zA-Z]+"))
                            .collect(Collectors.toList());
                    log.debug("words:{}", words1);
                    return words1.iterator();
                });

        JavaPairRDD<String, Integer> paris = words.mapToPair((PairFunction<String, String, Integer>) s -> new Tuple2<>(s, 1));

        JavaPairRDD<String, Integer> wcs = paris.reduceByKey((Function2<Integer, Integer, Integer>) Integer::sum);

        JavaPairRDD<Integer, String> sortedWcs = wcs.mapToPair((PairFunction<Tuple2<String, Integer>, Integer, String>) tuple2 -> new Tuple2<>(tuple2._2, tuple2._1)).sortByKey(false);

        sortedWcs.foreach((VoidFunction<Tuple2<Integer, String>>) tuple2 -> log.info("{} ... {}", tuple2._1, tuple2._2));

        sc.close();


    }

}
